package com.asiainfo.dacp.datastash.broker.redis.service;

import java.util.List;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.vo.Broker;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.flow.core.domain.Flow;
import com.asiainfo.dacp.datastash.broker.redis.RedisConstant;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;

@Service
public class RedisFlowService {
	
	@Resource(name="redisTemplate")
	private HashOperations<String, String, Flow> hash;

	@Resource(name="redisTemplate")
	private ListOperations<String, String> queue;
	
	public void createOrUpdate(Flow flow) {
		hash.put(RedisConstant.FLOW, flow.buildSerNo(), flow);
	}
	
	public Flow getFlow(String serNo){
		return hash.get(RedisConstant.FLOW, serNo);
	}

	public void putUnitWait(Flow flow) {
		queue.leftPush(RedisConstant.UNIT_WAIT_QUEUE, flow.buildSerNo());
	}

	public String fetchWaitUnit() {
		return queue.rightPop(RedisConstant.UNIT_WAIT_QUEUE);
	} 
	public void requeueUnitWait(Flow obj) {
		putUnitWait(obj);
	}

	public void putTasksWait(Flow flow) {
		queue.leftPush(RedisConstant.TASK_WAIT_QUEUE, flow.buildSerNo());
	}
	
	public String fetchTasksWait(){
		return queue.rightPop(RedisConstant.TASK_WAIT_QUEUE);
	}  
	
	public void putTaskWork(List<StashUnitTask> tasks) {
		for(StashUnitTask task: tasks){
			putTaskWork(task);
		}
	}
	
	public void putTaskWork( StashUnitTask task){
		String key = String.format(RedisConstant.BROKER_WORK_QUEUE_KEY, task.getBrokerId());
		queue.leftPush(key, task.buildSerNo());
	}
	
	public String fetchTaskWork(){
		return queue.rightPop(RedisConstant.BROKER_CURRENT_WORK_QUEUE_KEY);
	}

	

	
}
